/*********************************************************************
 * // 使用示例：
 * 
 *  #include <thread>
 *  
 *  #include <hude/base/log.hpp>
 *  #include <hude/base/hex.hpp>
 *  #include <hude/utils/parses.hpp>
 *  
 *  #include "test/test.hpp"
 *  
 *  struct case_parses_t : public case_t
 *  {
 *      enum : size_t { emCOUNT = 1000U, emINTERVAL = 1U, emTIMEOUT = 2U };
 *  
 *      //声明解析栈类型
 *      using parses_t = utils::parses_t< byte_t >;
 *  
 *      //为测试构造一个循环数据管道
 *      struct loop_t
 *      {
 *          parses_t  parses_;
 *  
 *          bool operator()( const parses_t::buffer_t& buffer )
 *          {
 *              for( auto byte : buffer )
 *              {
 *                  auto ptr = parses_.buffer_alloc( 1U );
 *                  (*ptr)[0] = byte;
 *                  parses_.buffer_push( ptr );
 *              }
 *  
 *              return true;
 *          }
 *      };
 *  
 *      // 声明同步请求类型, 同步请求对象属于解析栈的订阅者
 *      using observe_t = utils::sync_observe_t< loop_t, parses_t >; 
 *  
 *      // 测试过程
 *      bool run() override
 *      {
 *          const size_t lenght = 8U;
 *  
 *          size_t ret = 0, dop = 0, to = 0;
 *  
 *          // 测试数据的模板
 *          byte_t content[lenght + 2] = { 0xaa, 0xbb, 0xcc, 0xdd, 0U, 0U, 0U, 0U, 0x12, 0x34 };
 *  
 *          // 循环数据管道对象
 *          loop_t loop;
 *  
 *          // 解析栈工作线程
 *          auto th_ = std::thread{ [&loop, &dop]()
 *          {
 *              // 桢查找器, 从祼数据中扫描一个桢结构的方法
 *              auto finder = []( parses_t::citr_t& b, parses_t::citr_t& e ) -> parses_t::citr_t
 *              {
 *                  while( b != e )
 *                  {
 *                      if( *b == 0xaa )
 *                      {
 *                          if( e - b < lenght )
 *                          {
 *                              return e;
 *                          }
 *  
 *                          const byte_t* h = &(*b);
 *  
 *                          if( h[1] == 0xbb && h[2] == 0xcc && h[3] == 0xdd )
 *                          {
 *                              parses_t::citr_t r = b;
 *  
 *                              b += lenght;
 *  
 *                              return r;
 *                          }
 *                      }
 *  
 *                      ++b;
 *                  }
 *  
 *                  return ( b = e );
 *              };
 *  
 *              // 无效数据丢弃事件
 *              auto drop = [&dop]( parses_t::citr_t b, parses_t::citr_t e ) -> void
 *              {
 *                  dop += e - b;
 *  
 *                  log_war( "droped bytes: %u(%s)", e - b, hude::utils::to_hex( &(*b), e - b ).c_str() );
 *              };
 *  
 *              // 开始解析栈循环
 *              loop.parses_.process( finder, drop );
 *          } };
 *  
 *          // 一次同步请求
 *          auto request = [&ret, &to]( size_t index, loop_t& loop, const parses_t::buffer_t& buffer )
 *          {
 *              // 声明请求对象，请求对象是一次性的，不可重用。
 *              observe_t observe{ &loop, &loop.parses_ };
 *  
 *              // 请求回应的桢处理回调
 *              auto on_reply = [index]( parses_t::citr_t b, parses_t::citr_t e ) -> observe_t::result_t
 *              {
 *                  const uint32_t* p = (const uint32_t*)&(*(b + 4));
 *  
 *                  if( *p == index )
 *                  {
 *                      return observe_t::result_t::emRET_SUCCESS;
 *                  } else {
 *                      return observe_t::result_t::emRET_FRAME;
 *                  }
 *              };
 *  
 *              // 发起请求
 *              auto err = observe.request( buffer, on_reply, emTIMEOUT );
 *  
 *              static const char* txt[] =
 *              {
 *                  "emRET_SUCCESS",
 *                  "emRET_WAITTING",
 *                  "emRET_SEND",
 *                  "emRET_FRAME",
 *                  "emRET_TIMEOUT",
 *                  "emRET_CANCEL"
 *              };
 *  
 *              if( err == observe_t::result_t::emRET_SUCCESS )
 *              {
 *                  log_out( "sync request %u OK", index );
 *                  ret += 1;
 *              } else {
 *                  log_err( "sync request %u result: %s", index, txt[(int)err] );
 *                  to += 1;
 *              }
 *          };
 *  
 *          // 重复发起同步请求
 *          {
 *              for( uint32_t i = 0; i < emCOUNT; ++i )
 *              {
 *                  log_out( "push a packet: %u", i );
 *  
 *                  auto* p = (uint32_t*)(content + 4);
 *                  *p = i;
 *  
 *                  request( i, loop, parses_t::buffer_t{ content, content + sizeof( content ) } );
 *  
 *                  std::this_thread::sleep_for( std::chrono::milliseconds( emINTERVAL ) );
 *              }
 *          }
 *  
 *          // 测试结束清理工作
 *          {
 *              std::this_thread::sleep_for( std::chrono::milliseconds( 100U ) );
 *  
 *              loop.parses_.cannel();
 *  
 *              th_.join();
 *  
 *              log_nte( "=======================================================================" );
 *  
 *              if( to )
 *              {
 *                  log_war( "timeout count: %u / %u", to, emCOUNT );
 *              }
 *  
 *              if( ret < emCOUNT )
 *              {
 *                  log_war( "process count not full: %u / %u", ret, emCOUNT );
 *              }
 *  
 *              if( dop != emCOUNT * 2 )
 *              {
 *                  log_err( "drop bytes count error: %u / %u", dop, emCOUNT * 2 );
 *              }
 *          }
 *  
 *          return ( ( to + ret ) == emCOUNT ) && dop == emCOUNT * 2;
 *      }
 *  };
 * 
 ***************************************************************************************/

#ifndef __parses_kenny_kfjiweur89u83r33e3e__
#define __parses_kenny_kfjiweur89u83r33e3e__

#include <map>
#include <vector>
#include <memory>
#include <atomic>
#include <mutex>
#include <future>
#include <cassert>
#include <functional>

#include <hude/base/block_queue.hpp>
#include <hude/base/base_type.hpp>
#include <hude/base/optional.hpp>
#include <hude/base/log.hpp>

namespace hude
{
    namespace utils
    {
        /*********************************************************************
         * @brief 同步请求对象，提供请求结果的同步操作
         */
        template< typename _sender, typename _parses >
        class sync_observe_t
        {
        public:
            enum result_t
            {
                emRET_SUCCESS,
                emRET_SEND,
                emRET_REQUEST,
                emRET_FRAME,
                emRET_FINISH,
                emRET_WAITTING,
                emRET_PROCESSED,
                emRET_UNKNOWN,
                emRET_TIMEOUT,
                emRET_CANCEL
            };

            using sender_t     = _sender;
            using parses_t     = _parses;
            using this_t       = sync_observe_t< sender_t, parses_t >;
            using request_t    = typename parses_t::buffer_t;
            using citr_t       = typename parses_t::citr_t;
            using promise_t    = std::promise< result_t >;
            using future_t     = std::future< result_t >;
            using on_reply_t   = std::function< result_t( citr_t, citr_t ) >;

        public:
            /*********************************************************************
             * @brief 构造同步请求
             * @param sender_t 发送器， 指向可发送数据桢的对象的指针
             * @param parses_t 解析栈， 指向目标解析栈对象的指针
             */
            sync_observe_t( sender_t* sender, parses_t* parses ) : sender_( sender ), parses_( parses ) {}

            /*********************************************************************
             * @brief 发起同步请求
             * @param request_t 请求桢的数据缓存
             * @param on_reply_t 请求回应的回调处理， 可处理多桢数据:
             *                      - 当处理中间桢时需要返回emRET_WAITTING
             *                      - 完尾桢数据时需要返回emRET_SUCCESS
             *                      - 桢错误返回emRET_FRAME
             * @param duration 请求的超时时间, parses_t::FOREVER为从不超时
             * @param bool 是否自动清除订阅列表项
             */
            result_t request( const request_t&, on_reply_t, std::chrono::system_clock::duration, bool = true );

            /*********************************************************************
             * @brief 发起同步请求
             * @param request_t 请求桢的数据缓存
             * @param on_reply_t 请求回应的回调处理， 可处理多桢数据:
             *                      - 当处理中间桢时需要返回emRET_WAITTING
             *                      - 完尾桢数据时需要返回emRET_SUCCESS
             *                      - 桢错误返回emRET_FRAME
             * @param size_t 请求的超时时间, 0U为从不超时
             * @param bool 是否自动清除订阅列表项
             */
            result_t request( const request_t&, on_reply_t, size_t = 500U, bool = true );

            /*********************************************************************
             * @brief 取消当前的同步请求
             */
            void cancel();

        private:
            sync_observe_t( sync_observe_t&& ) = delete;
            sync_observe_t( const sync_observe_t& ) = delete;
            sync_observe_t& operator=( const sync_observe_t& ) = delete;

        private:
            sender_t    *sender_;
            parses_t    *parses_;
            promise_t   promise_;

        };

        /*********************************************************************
         * @brief 小于比较符算子
         */
        template< typename _value >
        struct _less_t { inline bool operator()( const _value& a, const _value& b ) const { return a < b; } };

        /*********************************************************************
         * @brief 通用的二进制解析栈，提供数据桢识别接口和数据桢的订阅者机制
         */
        template< typename _value = uint8_t, size_t _max = 0U >
        class parses_t
        {
        public:
            enum : size_t { emMAX_POLL = _max, emMAX_NOT = 0U };

            typedef _value                                             value_t;
            typedef parses_t< _value, _max >                           this_t;

            typedef std::chrono::system_clock                          clock_t;
            typedef typename clock_t::duration                         duration_t;
            typedef typename clock_t::time_point                       timepoint_t;

            typedef std::vector< value_t >                             buffer_t;
            typedef std::shared_ptr< buffer_t >                        item_t;
            typedef utils::block_queue_t< item_t >                     poll_t;
            typedef typename buffer_t::const_iterator                  citr_t;
            typedef std::function< citr_t( citr_t&, citr_t& ) >        finder_t;

            typedef std::function< void() >                            idle_t;
            typedef std::function< void( citr_t, citr_t ) >            drop_t;
            typedef std::function< bool( citr_t&, citr_t&, bool& ) >   observer_t;
            typedef _less_t< timepoint_t >                             less_t;
            typedef std::multimap< timepoint_t, observer_t, less_t >   queue_t;
            typedef typename queue_t::const_iterator                   oid_t;
            typedef std::mutex                                         mutex_t;
            typedef std::atomic< bool >                                running_t;

        public:
            static const duration_t  FOREVER;
            static const timepoint_t TP_FOREVER;

        public:
            parses_t() : running_( false ) {}
            ~parses_t() {}

            /*********************************************************************
             * @brief 增加数据桢的订阅者, 并指定其超时，超时到达后该订阅者自动删除
             * @param 订阅者
             * @param 订阅者超时
             * @return 订阅者id，后继可用于删除该订阅者
             */
            oid_t   observer_add( observer_t, duration_t );

            /*********************************************************************
             * @brief 增加数据桢的订阅者, 并指定其超时，超时到达后该订阅者自动删除
             * @param 订阅者
             * @param 订阅者超时
             * @return 订阅者id，后继可用于删除该订阅者
             */
            oid_t   observer_add( observer_t, size_t = 3000U );

            /*********************************************************************
             * @brief 删除数据桢的订阅者
             * @param 订阅者id
             */
            void    observer_del( oid_t );

            /*********************************************************************
             * @brief 删除所有数据桢的订阅者
             * @param 订阅者id
             */
            void    observer_clr();

            /*********************************************************************
             * @brief 订阅者数量
             */
            size_t  observer_siz() const;

            /*********************************************************************
             * @brief 设置数据桢的预处理订阅者，所有数据桢都会由该订阅者先处理
             */
            inline  void observer_pre( observer_t obs ) { pre_ = obs; }

            /*********************************************************************
             * @brief 设置数据桢的收尾处理订阅者，如果数据桢没有相关订阅者（预处理订阅者不算）处理，则最后由该订阅者处理
             */
            inline  void observer_trm( observer_t obs ) { trm_ = obs; }

            /*********************************************************************
             * @brief 申请一个缓冲区
             */
            item_t  buffer_alloc();
            item_t  buffer_alloc( size_t );
            template< typename _itr >
            item_t  buffer_alloc( _itr, _itr );

            /*********************************************************************
             * @brief 释放一个缓冲区
             */
            void  buffer_free( item_t& );

            /*********************************************************************
             * @brief 压入需要处理的原始数据, 该原始数据为一个buffer的引用计数对象
             */
            bool    buffer_push( item_t );

            /*********************************************************************
             * @brief 压入需要处理的原始数据, 该原始数据为原始缓冲及长度
             */
            bool    buffer_push( const void*, size_t );

            /*********************************************************************
             * @brief 压入需要处理的原始数据, 该原始数据为一对迭代器
             */
            template< typename _itr >
            bool    buffer_push( _itr, _itr );

            /*********************************************************************
             * @brief 压入需要处理的原始数据, 该原始数据为缓冲区类型
             */
            bool    buffer_push( hude::buffer_t&& );

            /*********************************************************************
             * @brief 压入需要处理的原始数据, 该原始数据为缓冲区类型
             */
            bool    buffer_push( const hude::buffer_t& );

            /*********************************************************************
             * @brief 待处理的原始数据buffer的数量
             */
            size_t  buffer_siz() const;

            /*********************************************************************
             * @brief 解析栈的一次解析处理
             * @param finder_t 数据桢的扫描器，用于从原始数据顶部查找和识别一个数据桢
             * @param drop_t 无效数据丢弃事件，当有无效数据被丢弃时触发事件
             * @param ms 阻塞的超时时间
             * @return 返回是否处理了有效数据
             */
            bool parse( finder_t finder, drop_t drop = drop_t{}, hude::optional_t< std::chrono::milliseconds > = hude::optional_null_t{} );

            /*********************************************************************
             * @brief 解析栈的处理一轮循环，直到缓存中没有完整的桢
             * @param finder_t 数据桢的扫描器，用于从原始数据顶部查找和识别一个数据桢
             * @param drop_t 无效数据丢弃事件，当有无效数据被丢弃时触发事件
             * @param ms 阻塞的超时时间
             * @param bool 是否自动清除订阅列表项
             */
            void parse_all( finder_t finder, drop_t drop = drop_t{}, hude::optional_t< std::chrono::milliseconds > = hude::optional_null_t{}, bool = false );

            /*********************************************************************
             * @brief 解析栈的处理循环
             * @param finder_t 数据桢的扫描器，用于从原始数据顶部查找和识别一个数据桢
             * @param drop_t 无效数据丢弃事件，当有无效数据被丢弃时触发事件
             * @param idle_t 等待超时后的闲置事件回调
             * @param size_t 单次检测超时
             */
            void process( finder_t, drop_t = drop_t{}, idle_t = idle_t{}, std::size_t = 100U );

            /*********************************************************************
             * @brief 取消解析栈的处理循环
             */
            void cannel() { running_ = false; }

        private:
            /*********************************************************************
             * @brief 在顶部压入数据桢识别未处理的剩余原始数据
             */
            void    buffer_head( citr_t, citr_t );

            duration_t    time_next( timepoint_t );
            void          time_out( timepoint_t );
            void          packet( citr_t b, citr_t e );

        private:
            std::atomic< bool > running_;
            mutable mutex_t  mutex_;

            observer_t pre_, trm_;

            poll_t   free_;
            poll_t   poll_;
            queue_t  queue_;
            buffer_t wait_;


        private:
            parses_t( parses_t&& ) = delete;
            parses_t( const parses_t& ) = delete;
            parses_t& operator=( const parses_t& ) = delete;

        };

        //--------------------------------------------------------------------------------
        template< typename _sender, typename _parses >
        inline typename sync_observe_t< _sender, _parses >::result_t
            sync_observe_t< _sender, _parses >::request( const request_t& req, on_reply_t res, size_t timeout, bool update )
        {
            return this->request( req, res, std::chrono::milliseconds( timeout ), update );
        }

        //--------------------------------------------------------------------------------
        template< typename _sender, typename _parses >
        typename sync_observe_t< _sender, _parses >::result_t
            sync_observe_t< _sender, _parses >::request( const request_t& req, on_reply_t res,
                typename std::chrono::system_clock::duration timeout, bool update )
        {
            assert( sender_ != nullptr );
            assert( parses_ != nullptr );
            assert( !! res );

            auto obs = [=]( citr_t& b, citr_t& e, bool& c ) -> bool
            {
                if( c ) // if timeout
                {
                    promise_.set_value( result_t::emRET_TIMEOUT );

                    return true;
                } else {
                    auto ret = res( b, e );

                    if( ret < result_t::emRET_FINISH )
                    {
                        c = true;

                        promise_.set_value( ret );
                    }

                    return ( ret <= result_t::emRET_PROCESSED );
                }

                return false;
            };

            promise_ = promise_t{};

            std::future< result_t > future  = promise_.get_future();

            auto oid = parses_->observer_add( obs, timeout );

            bool ret = (*sender_)( req );

            if( ! ret ) return result_t::emRET_SEND;

            std::future_status state = std::future_status::ready;

            if( timeout == std::chrono::system_clock::duration{} )
            {
                future.wait();
            } else {
                state = future.wait_for( timeout );
            }

            if( state == std::future_status::timeout )
            {
                if( ! update ) parses_->observer_del( oid );
                return result_t::emRET_TIMEOUT;
            } else {
                return future.get();
            }
        }

        //--------------------------------------------------------------------------------
        template< typename _sender, typename _parses >
        void sync_observe_t< _sender, _parses >::cancel()
        {
            promise_.set_value( result_t::emRET_CANCEL );
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        const typename parses_t< _value, _max >::duration_t parses_t< _value, _max >::FOREVER =
            parses_t< _value, _max >::duration_t{};

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        const typename parses_t< _value, _max >::timepoint_t parses_t< _value, _max >::TP_FOREVER =
            parses_t< _value, _max >::timepoint_t{};

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        inline typename parses_t< _value, _max >::oid_t   parses_t< _value, _max >::observer_add( observer_t obs, size_t timeout )
        {
            return this->observer_add( obs, std::chrono::milliseconds( timeout ) );
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        typename parses_t< _value, _max >::oid_t   parses_t< _value, _max >::observer_add(
            observer_t obs, duration_t timeout )
        {
            std::lock_guard< mutex_t > _( mutex_ );

            auto ret = queue_.emplace( ( timeout == FOREVER )? TP_FOREVER : clock_t::now() + timeout, obs );
            return ret;
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        void    parses_t< _value, _max >::observer_del( oid_t oid )
        {
            std::lock_guard< mutex_t > _( mutex_ );

            queue_.erase( oid );
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        void    parses_t< _value, _max >::observer_clr()
        {
            std::lock_guard< mutex_t > _( mutex_ );

            bool remove;
            citr_t none;

            for( auto itr = queue_.begin(); itr != queue_.end(); ++itr )
            {
                itr->second( none, none, ( remove = true ) );
            }

            queue_.clear();
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        size_t    parses_t< _value, _max >::observer_siz() const
        {
            std::lock_guard< mutex_t > _( mutex_ );

            return queue_.size();
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        bool    parses_t< _value, _max >::buffer_push( item_t item )
        {
            if( emMAX_POLL == emMAX_NOT || poll_.size() < emMAX_POLL )
            {
                poll_.push( item );

                return true;
            } else {
                return false;
            }
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        bool    parses_t< _value, _max >::buffer_push( const void* p, size_t n )
        {
            auto b = static_cast< const uint8_t* >( p );
            return buffer_push( buffer_alloc( b, b + n ) );
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        template< typename _itr >
        bool    parses_t< _value, _max >::buffer_push( _itr b, _itr e )
        {
            return buffer_push( buffer_alloc( b, e ) );
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        bool    parses_t< _value, _max >::buffer_push( hude::buffer_t&& buffer )
        {
            auto tmp = buffer_alloc();
            std::swap( *tmp, buffer );
            return buffer_push( tmp );
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        bool    parses_t< _value, _max >::buffer_push( const hude::buffer_t& buffer )
        {
            return buffer_push( buffer_alloc( buffer.cbegin(), buffer.cend() ) );
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        void    parses_t< _value, _max >::buffer_head( citr_t b, citr_t e )
        {
            typename poll_t::operator_t op( &poll_ );

            auto item = buffer_alloc( b, e );

            if( ! op->empty() )
            {
                auto f = op->front();
                op->pop_front();

                std::copy( f->begin(), f->end(), std::back_inserter( *item ) );
            }

            op->push_front( item );
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        size_t    parses_t< _value, _max >::buffer_siz() const
        {
            return poll_.size();
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        void    parses_t< _value, _max >::packet( citr_t b, citr_t e )
        {
            bool remove = false;

            if( pre_ && pre_( b, e, remove ) )
            {
                return;
            } else {
                std::lock_guard< mutex_t > _( mutex_ );

                for( typename queue_t::iterator itr = queue_.begin(); itr != queue_.end(); )
                {
                    bool stop = itr->second( b, e, ( remove = false ) );

                    if( remove )
                    {
                        itr = queue_.erase( itr );
                    } else {
                        ++itr;
                    }

                    if( stop )
                    {
                        return;
                    }
                }
            }

            if( trm_ )
            {
                trm_( b, e, ( remove = false ) );
            }
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        typename parses_t< _value, _max >::item_t    parses_t< _value, _max >::buffer_alloc( size_t siz )
        {
            auto ret = buffer_alloc();

            ret->resize( siz );

            return ret;
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        template< typename _itr >
        typename parses_t< _value, _max >::item_t    parses_t< _value, _max >::buffer_alloc( _itr b, _itr e )
        {
            auto ret = buffer_alloc();

            ret->assign( b, e );

            return ret;
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        typename parses_t< _value, _max >::item_t    parses_t< _value, _max >::buffer_alloc()
        {
            if( free_.empty() )
            {
                free_.push( item_t( new buffer_t ) );
                free_.push( item_t( new buffer_t ) );
            }

            auto ret = free_.front();
            free_.pop();

            return ret;
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        void    parses_t< _value, _max >::buffer_free( item_t& item )
        {
            item->clear();
            free_.push( item );
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        typename parses_t< _value, _max >::duration_t    parses_t< _value, _max >::time_next( timepoint_t now )
        {
            std::lock_guard< mutex_t > _( mutex_ );

            if( queue_.empty() )
            {
                return duration_t( std::chrono::duration_cast< duration_t >( std::chrono::milliseconds( 500U ) ) );
            } else {
                auto head = queue_.begin();

                return ( head->first > now )? head->first - now : duration_t( 1U );
            }
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        void    parses_t< _value, _max >::time_out( timepoint_t now )
        {
            std::lock_guard< mutex_t > _( mutex_ );

            bool remove;
            citr_t none;

            for( auto itr = queue_.begin(); itr != queue_.end(); )
            {
                if( itr->first != TP_FOREVER && itr->first < now )
                {
                    // log_dug( "timeout item." );

                    itr->second( none, none, ( remove = true ) );

                    log_war( "timeout %p from %p", &(itr->second), this );

                    itr = queue_.erase( itr );
                } else {
                    ++itr;
                    break;
                }
            }
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        bool parses_t< _value, _max >::parse( finder_t finder, drop_t drop, hude::optional_t< std::chrono::milliseconds > ms )
        {
            auto cb = [=]( item_t& item ) -> void
            {
                std::copy( item->cbegin(), item->cend(), std::back_inserter( wait_ ) );

                buffer_free( item );

                parses_t::citr_t b = wait_.cbegin();
                parses_t::citr_t e = wait_.cend();

                while( b < e )
                {
                    parses_t::citr_t t = b;

                    auto h = finder( b, e );

                    if( drop )
                    {
                        auto s = (h == e)? b : h;

                        if( t != s )
                        {
                            drop( t, s );
                        }
                    }

                    if( h != e )
                    {
                        this->packet( h, b );
                    } else {
                        break;
                    }
                }

                if( b < e )
                {
                    // log_dug( "\t swap buffer( %d, %d )", e - b, e - wait_.cbegin() );

                    buffer_t tmp{ b, e };
                    wait_.swap( tmp );
                } else {
                    wait_.clear();
                }
            };

            return poll_.consume( cb, true, ms );
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        void parses_t< _value, _max >::parse_all( finder_t finder, drop_t drop,
            hude::optional_t< std::chrono::milliseconds > ms, bool update )
        {
            if( ! finder ) return;

            bool state;

            do
            {
                auto now = clock_t::now();

                if( update ) this->time_out( now );

                if( ms )
                {
                    auto tp = std::chrono::duration_cast<std::chrono::milliseconds>( time_next( now ) );

                    state = parse( finder, drop, ( tp == std::chrono::milliseconds() )? *ms : tp );
                } else {
                    state = parse( finder, drop, hude::optional_null_t{} );
                }
            }
            while( state );
        }

        //--------------------------------------------------------------------------------
        template< typename _value, size_t _max >
        void    parses_t< _value, _max >::process( finder_t finder, drop_t drop, idle_t idle, std::size_t timeout )
        {
            if( ! finder ) return;

            running_ = true;

            while( running_ )
            {
                parse_all( finder, drop, std::chrono::milliseconds( timeout ), true );

                if( idle ) idle();
            }
            
            observer_clr(); 
        }
    } // namespace utils
} // namespace nsPGL

#endif // __parses_kenny_kfjiweur89u83r33e3e__
